/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.reading;

import java.nio.ByteBuffer;

import edu.ucsd.db.bassarisk.data.api.IValueType.IValueReference;
import edu.ucsd.db.bassarisk.data.compression.VariableLengthUnsignedLongSequenceCodec;
import edu.ucsd.db.bassarisk.data.values.LongType;
import edu.ucsd.db.bassarisk.data.values.LongType.LongTools;
import edu.ucsd.db.bassarisk.data.values.LongType.ObjectLongReference;
import edu.ucsd.db.bassarisk.storage.common.ColumnStore;
import edu.ucsd.db.bassarisk.storage.common.IColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.loading.ColumnSegmentInfo;
import edu.ucsd.db.bassarisk.storage.reading.IColumnIndexReader.DrawerInfo;

public class VariableLengthLongDrawerIndexReader implements
    IDrawerIndexReader<LongType> {

  protected static class Cursor {
    protected final ObjectLongReference cardEntryDelta = new ObjectLongReference();
    protected int cardIndex;
    protected long cardStartEntry;
    protected int entryByteOffset;
  }

  private static final short drawerIndexVersion = 3;
  private static final int HEADER_BYTES = 5;
  private final Cursor current;
  private int dataCardCount;
  private ByteBuffer drawerBuffer;
  private IColumnSegmentInfo<LongType> drawerInfo;
  private final VariableLengthUnsignedLongSequenceCodec longCodec;

  public VariableLengthLongDrawerIndexReader() {
    this.longCodec = VariableLengthUnsignedLongSequenceCodec.INSTANCE;
    LongTools.INSTANCE.getGenericCodec();

    this.current = new Cursor();
  }

  // For use in test
  public VariableLengthLongDrawerIndexReader(final ByteBuffer injectedBuffer,
                                             final VariableLengthUnsignedLongSequenceCodec injectedCodec,
                                             final Cursor injectedCursor,
                                             final IColumnSegmentInfo<LongType> injectedDrawerInfo,
                                             final int injectedDataCardCount) {
    this.longCodec = injectedCodec;
    LongTools.INSTANCE.getGenericCodec();
    this.drawerBuffer = injectedBuffer;
    this.current = injectedCursor;
    this.drawerInfo = injectedDrawerInfo;
    this.dataCardCount = injectedDataCardCount;
  }

  @Override
  public void getCardData(final int cardIndex, final CardInfo<LongType> info) {
    assert cardIndex >= this.current.cardIndex;
    assert cardIndex < this.dataCardCount;

    while (cardIndex > this.current.cardIndex) {
      this.stepCursor();
    }
    info.reset(this.current.cardIndex, this.current.cardEntryDelta.getValue(),
               this.current.cardStartEntry);
  }

  @Override
  public boolean getCardDataForEntryIndex(final int targetEntryIndex,
                                          final CardInfo<LongType> info) {
    assert targetEntryIndex >= this.current.cardStartEntry; // forward only

    if (!this.drawerInfo.hasEntry(targetEntryIndex)) { return false; }

    while (targetEntryIndex >= (this.current.cardStartEntry + this.current.cardEntryDelta
        .getValue())) {
      this.stepCursor();
    }
    info.reset(this.current.cardIndex, this.current.cardEntryDelta.getValue(),
               this.current.cardStartEntry);
    return true;
  }

  @Override
  public boolean getCardDataForValue(final IValueReference<LongType> value,
                                     final CardInfo<LongType> info) {
    String message = "Operation is only supported on sorted columns";
    throw new UnsupportedOperationException(message);
  }

  @Override
  public int getDataCardCount() {
    return this.dataCardCount;
  }

  @Override
  public int getDrawerIndexVersion() {
    return drawerIndexVersion;
  }

  @Override
  public boolean getNextCardData(final CardInfo<LongType> info) {
    if ((this.current.cardIndex + 1) == this.dataCardCount) { return false; }
    if (this.drawerBuffer == null) { return false; } // HACK FIXME

    this.stepCursor();
    info.reset(this.current.cardIndex, this.current.cardEntryDelta.getValue(),
               this.current.cardStartEntry);
    return true;
  }

  @Override
  public CardInfo<LongType> makeCardInfoInstance() {
    return new CardInfo<LongType>(LongTools.INSTANCE);
  }

  @Override
  public void resetDrawer(final ByteBuffer drawerBuffer,
                          final ColumnSegmentInfo<LongType> drawerInfo) {
    DrawerInfo<LongType> di = (DrawerInfo<LongType>) drawerInfo;
    int drawerHeaderAddress = di.getDrawerHeaderPageIndex();
    this.drawerInfo = drawerInfo;
    this.drawerBuffer = drawerBuffer;

    // Extract header information from the tail of the drawerBuffer
    int headerPosition =
        ((drawerHeaderAddress + 1) * ColumnStore.BYTES_IN_CARD) - HEADER_BYTES;
    this.drawerBuffer.position(headerPosition);

    int indexCardCount = this.drawerBuffer.get();
    this.dataCardCount = this.drawerBuffer.getShort();
    assert drawerIndexVersion == this.drawerBuffer.getShort();
    this.drawerBuffer.clear();
    // Leave the drawerBuffer in a cleared state

    int firstIndexCard = drawerHeaderAddress - indexCardCount + 1;
    this.current.cardIndex = -1;
    this.current.cardStartEntry = 0;
    this.current.entryByteOffset = firstIndexCard * ColumnStore.BYTES_IN_CARD;
    this.current.entryByteOffset +=
        this.longCodec.decode(this.drawerBuffer, this.current.cardEntryDelta,
                              this.current.entryByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);

  }

  private void stepCursor() {
    this.current.cardStartEntry += this.current.cardEntryDelta.getValue();
    this.current.entryByteOffset +=
        this.longCodec.decode(this.drawerBuffer, this.current.cardEntryDelta,
                              this.current.entryByteOffset,
                              ColumnStore.BYTES_IN_DRAWER);
    this.current.cardIndex++;
  }
}
